package mysql.socket;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.util.internal.OutOfDirectMemoryError;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SystemPropertyUtil;
import org.apache.commons.lang.NotImplementedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.SocketAddress;

/**
 * @ClassName NettyCacheSocketChannel
 * @Description TODO
 * @Author fengk
 * @Date 2018/9/4 12:57
 */

public class NettyCacheSocketChannel implements SocketChannel {
    private static final Logger logger= LoggerFactory.getLogger(NettyCacheSocketChannel.class);

    private static final int WAIT_PERIOD=10;//每次操作等待
    private static final int DEFAULT_INIT_BUFFER_SIZE=1024*1024;//1K
    private static final int DEFAULT_MAX_BUFFER_SIZE=16* DEFAULT_INIT_BUFFER_SIZE;
    private Channel channel=null;
    private Object lock=new Object();
    private ByteBuf cache= PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE);
    private int maxDirectBuffer=cache.maxCapacity();

    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    public ByteBuf getCache() {
        return cache;
    }

    public void setCache(ByteBuf cache) {
        this.cache = cache;
    }

    public int getMaxDirectBuffer() {
        return maxDirectBuffer;
    }

    public void setMaxDirectBuffer(int maxDirectBuffer) {
        this.maxDirectBuffer = maxDirectBuffer;
    }

    public void writeCache(ByteBuf buf) throws InterruptedException,IOException{
        synchronized (lock){
            while(true){
                if(null==cache){
                    throw  new IOException("socket is closed !");
                }
                // buf is empty
                if(!buf.isReadable()){
                    //退出循环
                    break;
                }

                int readLen=buf.readableBytes();
                int deltaLen=readLen-cache.writableBytes();
                //读数大于可写
                if(deltaLen>0){
                    //避免频繁分配内存   避免频繁移动
                    if(cache.readerIndex()>=deltaLen){
                        //可以内存移动
                        //回收已读
                        cache.discardReadBytes();
                        //如果内存过大  收缩内存
                        int oldCacheCapacity=cache.capacity();
                        if(oldCacheCapacity>DEFAULT_MAX_BUFFER_SIZE){
                            //保留现有数据
                            int reserveCapacity=cache.writerIndex();
                            //reserveCapacity的计算结果是DEFAULT_INIT_BUFFER_SIZE的倍数
                            //对齐
                            //利用int的特性
                            reserveCapacity=((reserveCapacity-1)/DEFAULT_INIT_BUFFER_SIZE+1)*DEFAULT_INIT_BUFFER_SIZE;
                            int quarter=(reserveCapacity>>2);//除以4
                            // 对齐
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE;
                            reserveCapacity=reserveCapacity+quarter;
                            if(reserveCapacity<(oldCacheCapacity>>1)){
                                try{
                                    cache.capacity(reserveCapacity);
                                    logger.info("shrink cache : {} - {} = {} bytes",oldCacheCapacity,oldCacheCapacity-reserveCapacity,reserveCapacity);
                                }catch (OutOfMemoryError ignore){
                                    //以后不要超过当前大小
                                    setMaxDirectBuffer(oldCacheCapacity);
                                    logger.warn("system OutOfMemoryError: {} bytes",reserveCapacity,ignore);
                                }
                            }
                        }
                    }else {
                      //扩容
                        int oldCacheCapacity=cache.capacity();
                        //cache 上限限制
                        if(oldCacheCapacity<getMaxDirectBuffer()){
                            //至少扩大四分之一
                            //以DEFAULT_INIT_BUFFER_SIZE为最小粒度
                            int quarter=(oldCacheCapacity>>2);
                            quarter=((quarter-1)/DEFAULT_INIT_BUFFER_SIZE+1)*DEFAULT_INIT_BUFFER_SIZE;
                            //deltalen 的最小粒度是quarter
                            deltaLen=((deltaLen-1)/quarter+1)*quarter;//对齐
                            int newCacheCapacity=oldCacheCapacity+deltaLen;
                            if(newCacheCapacity>getMaxDirectBuffer()){
                                newCacheCapacity=getMaxDirectBuffer();
                            }
                            try{
                                cache.capacity(newCacheCapacity);
                                logger.info("expand cache : {} + {} = {} bytes",
                                        oldCacheCapacity,newCacheCapacity-oldCacheCapacity,
                                        newCacheCapacity);
                            }catch (OutOfDirectMemoryError e){
                                long maxDirectMemory= SystemPropertyUtil.getLong("io.netty.maxDirectMemory",-1);
                                if(maxDirectMemory<0){
                                    maxDirectMemory= PlatformDependent.maxDirectMemory();
                                }
                                if(getMaxDirectBuffer()>maxDirectMemory){
                                    setMaxDirectBuffer((int)maxDirectMemory);
                                    newCacheCapacity=getMaxDirectBuffer();
                                    logger.warn("resize maxDirectBuffer: {} bytes",getMaxDirectBuffer(),e);
                                    try{
                                        cache.capacity(newCacheCapacity);
                                        cache.capacity(newCacheCapacity);
                                        logger.info("expand cache : {} + {} = {} bytes",
                                                oldCacheCapacity,newCacheCapacity-oldCacheCapacity,
                                                newCacheCapacity);
                                    }catch (OutOfMemoryError ignore){
                                        setMaxDirectBuffer(oldCacheCapacity);//以后不要超过当前容量
                                        logger.warn("system OutOfMemoryError: {} bytes",newCacheCapacity,ignore);
                                    }
                                }else {
                                    //不是因为内存不够引起的
                                    //只能记录日志了

                                    setMaxDirectBuffer(oldCacheCapacity);
                                    logger.warn("cache OutOfDirectMemoryError: {} bytes", newCacheCapacity, e);
                                }
                            }catch (OutOfMemoryError ignore){
                                setMaxDirectBuffer(oldCacheCapacity);
                                logger.warn("cache OutOfMemoryError: {} bytes", newCacheCapacity, ignore);
                            }
                        }
                    }
                    //cache重新调整后
                    deltaLen=readLen-cache.writableBytes();
                }
                //if deltaLen == reaLen 那么cache.writableBytes = 0
                if (deltaLen!=readLen){
                    //deltaLen<=0 全部写入
                    //deltaSize>0 写入部分
                    if(deltaLen<=0){
                        cache.writeBytes(buf,readLen);
                        //全部写入  结束循环
                        break;
                    }else /*deltaLen>0*/{
                        cache.writeBytes(buf,readLen-deltaLen);//===== cache.writableBYtes
                    }
                }
                //cache 满
                //等待
                lock.wait(WAIT_PERIOD);
                // 回收已读空间，重置读写指针
                cache.discardReadBytes();
            }
        }
    }

    @Override
    public void write(byte[]... buf) throws IOException {
        if(channel!=null&&channel.isWritable()){
            channel.writeAndFlush(Unpooled.copiedBuffer(buf));
        }else {
            throw  new IOException("write failed ! channel has a problem! ");
        }
    }

    @Override
    public byte[] read(int readSize) throws IOException {
        return read(readSize,0);
    }

    @Override
    public byte[] read(int readSize, int timeout) throws IOException {
        int accumulatedWaitTime=0;
        //如果timeout > 0 自动生成超时时间 且读取长度超过DEFAULT_INIT_BUFFER_SIZE
        //timeout<=0 不生成超时时间
        // 且如果timeout > 0 自动生成超时时间 且读取长度不超过DEFAULT_INIT_BUFFER_SIZE 不生成超时时间
        if(timeout>0&&readSize>DEFAULT_INIT_BUFFER_SIZE){
            timeout*=(readSize/DEFAULT_INIT_BUFFER_SIZE+1);
        }
        do{
            if(readSize>cache.readableBytes()){
                if(null==channel){
                    throw  new IOException("socket has Interrupted !");
                }
                if(timeout>0){
                    accumulatedWaitTime+=WAIT_PERIOD;
                    if(accumulatedWaitTime>timeout){
                        StringBuilder sb=new StringBuilder("socket read timeout occured !");
                        sb.append(" readSize = ").append(readSize);
                        sb.append(" , readableBytes = ").append(cache.readableBytes());
                        sb.append(" ,timeout = ").append(timeout);
                        throw  new IOException(sb.toString());
                    }
                    synchronized (this){
                        try {
                            wait(WAIT_PERIOD);
                        }catch (InterruptedException e){
                            throw  new IOException("socket has Interrupted !");
                        }
                    }
                }
            }else {
                byte[] back=new byte[readSize];
                synchronized (lock){
                    cache.readBytes(back);
                }
                return back;
            }
        }while (true);
    }

    @Override
    public void read(byte[] data, int off, int len, int timeout) throws IOException {
        throw new NotImplementedException();
    }

    @Override
    public boolean isConnected() {
        return channel!=null?true:false;
    }

    @Override
    public SocketAddress getRemoteSocketAddress() {
        return channel!=null?channel.remoteAddress():null;
    }

    @Override
    public void close() {
        if(channel!=null){
            channel.close();
        }
        channel=null;
        synchronized (lock){
            cache.discardReadBytes();
            cache.release();
            cache=null;
        }
    }
}
